package com.bclz.task.request;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.MRConfig;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import com.bclz.config.JobRun;

public class RequeTimesStep2 {

	public static class RequestTimesMapStep2Task extends Mapper<LongWritable, Text, CountMap, NullWritable> {

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			// TODO Auto-generated method stub

			String[] str = value.toString().split("\t");
			context.write(new CountMap(str[0], Integer.valueOf(str[1])), NullWritable.get());
		}

	}

	public static class RequestTimesReduceStep2Task extends Reducer<CountMap, NullWritable, CountMap, NullWritable> {

		@Override
		protected void reduce(CountMap key, Iterable<NullWritable> values, Context result)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub

			result.write(key, NullWritable.get());

		}

	}

	public static void main(String[] args) {

		System.setProperty("HADOOP_USER_NAME", "hadoop");
		Configuration p = new Configuration();

		p.set("fs.defaultFS", "file:///");
		p.set(MRConfig.FRAMEWORK_NAME, "local");

		try {

			Job job = Job.getInstance(p);
			job.setJarByClass(RequeTimesStep2.class);

			job.setMapperClass(RequestTimesMapStep2Task.class);
			job.setReducerClass(RequestTimesReduceStep2Task.class);

			job.setMapOutputKeyClass(CountMap.class);
			job.setMapOutputValueClass(NullWritable.class);

			job.setOutputKeyClass(CountMap.class);
			job.setOutputValueClass(NullWritable.class);
			JobRun.runJob(job, "E:\\hadoop_input\\test\\output", "E:/hadoop_input/test/Compareoutput", 1);

		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

}
